package com.asap.demo.kafkastream;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class ReceveFromKafka {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Properties props = new Properties();
		props.put("bootstrap.servers", "192.168.1.25:9093");
		props.put("zookeeper.connect", "192.168.1.25:1813");
		props.put("group.id", "metric-group");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //key 反序列化
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("auto.offset.reset", "latest"); //value 反序列化

		DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer<String>(
				"flink_pressure_test",  //kafka topic
				new SimpleStringSchema(),  // String 序列化
				props)).setParallelism(1);

		dataStreamSource.print(); //把从 kafka 读取到的数据打印在控制台

		env.execute("Flink add data source");
	}
}
